package com.ltu.service.mqtt;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.ltu.config.mqtt.MqttConfiguration;
import com.ltu.util.SpringUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 生产消息
 * @author Administrator
 */
@Slf4j
@Component
public class MqttPushClient {
	@Autowired
	private MqttConfiguration mqttConfiguration;
	
    @Autowired
    private  MqttClient mqttClient;

    /**
     * 发布，默认qos为0，非持久化
     * @param topic
     * @param pushMessage
     */
    
    public void pushlish(String topic,String pushMessage){
    	log.info( String.valueOf(mqttClient.isConnected()) );
    	if(!mqttClient.isConnected()) {
    		 MqttListenterTopic mqttListenterTopic  = SpringUtil.getBean(MqttListenterTopic.class);
    		mqttConfiguration.connect(mqttListenterTopic);
    	}
        pushlish(0,false,topic,pushMessage);
    }

    /**
     * 发布
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */
    private void pushlish(int qos,boolean retained,String topic,String pushMessage){
        MqttMessage message=new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mqttTopic=  mqttClient.getTopic(topic);
        if(null== mqttTopic){
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token=mqttTopic.publish(message);
            token.waitForCompletion();
        }catch (MqttPersistenceException e){
            e.printStackTrace();
        }catch (MqttException e){
            e.printStackTrace();
        }

    }
    
    public  void  subscribe(String topic) {
    	try {
    		mqttClient.subscribe(topic,0);
    		
		} catch (MqttException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    	
    }
    
    public  void  subscribe(String topic,int qos) {
    	try {
    		mqttClient.subscribe(topic,qos);
    	} catch (MqttException e) {
    		// TODO Auto-generated catch block
    		e.printStackTrace();
    	}
    	
    }
    
    /**
     * 取消订阅某个主题
     *
     * @param topic
     */
    public void unsubscribe(String topic) {
        log.info("==============取消订阅主题==============" + topic);
        try {
        	mqttClient.unsubscribe(topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

}